package com.db.stockexchange.app;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import com.db.stockexchange.domain.Exchange;
import com.db.stockexchange.domain.ExecutionReport;
import com.db.stockexchange.domain.order.Order;
import com.db.stockexchange.network.RequestListener;
import com.db.stockexchange.network.Server.Connection;
import com.db.stockexchange.proto.Converter;
import com.db.stockexchange.proto.Protocol.ExchangeMessage;
import com.db.stockexchange.proto.Protocol.ExchangeMessage.MessageType;
import com.db.stockexchange.proto.ValidationException;
import com.google.protobuf.InvalidProtocolBufferException;

public class Dispatcher implements RequestListener {

	private static final int DEFAULT_POOL_SIZE = 4;
	
	private Map<Connection, String> connectionToUsername = new HashMap<>();
	private Set<String> activeUsers = new HashSet<>();
	
	private Executor pool = Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
	private Exchange exchange;
	
	private Logger logger = Logger.getLogger(Dispatcher.class.getName());
	
	public Dispatcher(Exchange exchange) {
		this.exchange = exchange;
	}

	@Override
	public void requestReceived(byte[] message, Connection connection) {
		pool.execute(new RequestHandler(message, connection));
	}
	
	private class RequestHandler implements Runnable {

		private byte[] message;
		private Connection connection;
		
		public RequestHandler(byte[] message, Connection connection) {
			this.message = message;
			this.connection = connection;
		}

		@Override
		public void run() {
			try {
				ExchangeMessage exchangeMessage = ExchangeMessage.parseFrom(message);
//				logger.info("NEW MESSAGE: " + exchangeMessage);
				
				if (!exchangeMessage.hasMessageType()) {
					sendReject("message type must be specified");
					return;
				}
				
				if (!connectionToUsername.containsKey(connection)) {
					if (exchangeMessage.getMessageType() != ExchangeMessage.MessageType.LOGON) {
						sendReject("logon required");
						return;
					}
				}
				
				switch (exchangeMessage.getMessageType()) {
				case NEW_ORDER:
					Order order = Converter.messageToOrder(exchangeMessage);
					order.setConnection(connection);
					order.setUsername(connectionToUsername.get(connection));
					
					for (ExecutionReport report : exchange.placeOrder(order)) {
						ExchangeMessage reportMessage = Converter.execReportToMessage(report);
						try {
							report.getConnection().write(reportMessage.toByteArray());
						} catch (IOException e) {
							logger.warning("failed to send execution report: " + e.getMessage());
						}
					}
					
					break;
					
				case LOGON:
					if (!exchangeMessage.hasUsername())
						sendReject("username not specified");
					else if (activeUsers.contains(exchangeMessage.getUsername()))
						sendReject("username already in use");
					else if ("".equals(exchangeMessage.getUsername())) {
						sendReject("username can't be empty");
					}
					else {
						logger.info("logon successful, user: " + exchangeMessage.getUsername());
						connectionToUsername.put(connection, exchangeMessage.getUsername());
						activeUsers.add(exchangeMessage.getUsername());
						try {
							connection.write(message);
						} catch (IOException ignore) {}
					}
					break;
					
				default:
					sendReject("unsupported message type");
					break;
				}
				
			} catch (InvalidProtocolBufferException e) {
				sendReject("unknown format");
			} catch (ValidationException e) {
				sendReject(e.getMessage());
			}
		}
		
		private void sendReject(String reason) {
			logger.info("sending reject: " + reason);
			
			ExchangeMessage rejectMessage = ExchangeMessage.newBuilder()
					.setMessageType(MessageType.REJECT)
					.setComment(reason)
					.build();
			try {
				connection.write(rejectMessage.toByteArray());
			} catch (IOException e) {
				logger.warning("failed to send reject: " + e.getMessage());
			}
		}

	}

}
